package top.zhangjianyong.test.reptiles;

import lombok.extern.slf4j.Slf4j;
import top.zhangjianyong.entity.ReptileUrl;
import top.zhangjianyong.thread.ReptileCallable;
import top.zhangjianyong.utils.JDBCUtil;
import top.zhangjianyong.utils.MailUtil;
import top.zhangjianyong.utils.ReptileContent;


import java.sql.SQLException;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.*;

/**
 * @author ： zhangjianyong
 * @date ： 2023/3/27 4:08 PM
 */
@Slf4j
public class Main {


    private static ThreadPoolExecutor pool;

    public static void init(String[] args) throws SQLException {
        int length = args.length;
        if (length > 6) {
            String host = args[0];
            String port = args[1];
            String dbname = args[2];
            String url = "jdbc:mysql://" + host + ":" + port + "/" + dbname + "?useSSL=false&allowMultiQueries=true&useUnicode=true&characterEncoding=UTF-8&serverTimezone=Hongkong";
            String user = args[3];
            String password = args[4];
            JDBCUtil.init(url, user, password);

            int poolSize = Integer.parseInt(args[5]);
            int queueSize = Integer.parseInt(args[6]);
            log.info("线程数：{}", poolSize);
            log.info("线程队列长度：{}", queueSize);
            QUEUE_SIZE = queueSize;
            pool = new ThreadPoolExecutor(poolSize, poolSize, 5, TimeUnit.SECONDS,
                    new LinkedBlockingDeque<>(queueSize), Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.AbortPolicy());
        } else {
            throw new RuntimeException("参数错误！！！");
        }
        if (ReptileContent.isEmptyUrl()) {
            throw new RuntimeException("当前数据库中地址为空！！！");
        }
    }

    private final static Queue<Future<Integer>> FUTURE_QUEUE = new LinkedList<>();
    private final static Queue<Future<Integer>> TMP_QUEUE = new LinkedList<>();
    private static Integer QUEUE_SIZE;

    public static void main(String[] args) throws Exception {
        int count = 0;


        try {
            init(args);//初始化数据库线程配置
            ReptileContent.init();//等待处理中数据回原始表
            //1.循环从数据库中取出网页地址
            while (true) {
                if (ReptileContent.isWait(QUEUE_SIZE / 2)) {//待处理表还有数据，等待3秒
                    Thread.currentThread().sleep(3000);
                    continue;
                }
                List<ReptileUrl> urls = ReptileContent.pollUrl(QUEUE_SIZE);
                for (ReptileUrl url : urls) {
                    /*去重复*/
                    if (ReptileContent.exists(url)) {
                        ReptileContent.del(url.getId());
                        continue;
                    }
                    ReptileContent.waitUrl(url);

                    if (!submitTask(url)) {
                        //线程满了跳出循环重新获取
                        break;
                    }
                    count++;
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            log.info("任务结束");
            //ReptileContent.save();
            MailUtil.send("爬虫任务结束", "爬取网页数量：" + count);
        }
    }

    public static boolean submitTask(ReptileUrl url) throws SQLException, ExecutionException, InterruptedException {
        try {
            log.info("加入处理线程：{}", url);
            pool.submit(new ReptileCallable(url));
            return true;
        } catch (RejectedExecutionException e) {
            //log.error(e.getMessage(), e);
            ReptileContent.initUrl(url);
            log.info("队列满了等待线程处理");
            waite();
            return false;
        }
    }

    /**
     * 等待一半的任务完成
     *
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void waite() throws ExecutionException, InterruptedException {
        Thread.currentThread().sleep(5000);
        int size = pool.getQueue().size();
        int n = QUEUE_SIZE * 4 / 5;
        log.info("[主线程等待]当前队列个数：{}", size);
        if (size > n) {
            waite();
        }
    }

}
